package com.jch.dm6_active_object.example;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

// 角色： ActiveObject.Proxy
public class AsyncRequestPersistence implements RequestPersistence {

    private static final long ONE_MINUTE_IN_SECONDS = 60;

    final static Logger log = LogManager.getLogger(AsyncRequestPersistence.class);

    final AtomicLong taskTimeConsumedPerInterval = new AtomicLong(0);
    final AtomicInteger requestSubmittedPerInterval = new AtomicInteger(0);

    // 角色： ActiveObject.Servant
    private final DiskBasedRequestPersistence delegate = new DiskBasedRequestPersistence();

    // 角色：ActiveObject.Scheduler
    private final ThreadPoolExecutor scheduler;

    private static class InstanceHolder {
        final static RequestPersistence INSTANCE = new AsyncRequestPersistence();
    }

    private AsyncRequestPersistence() {
        scheduler = new ThreadPoolExecutor(1, 3, 60 * ONE_MINUTE_IN_SECONDS, TimeUnit.SECONDS,
                // 角色：ActiveObject.ActivationQueue
                new ArrayBlockingQueue<Runnable>(200), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread;
                thread = new Thread(r, "AsyncRequestPersistence");
                return thread;
            }
        });
        scheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        Timer monitorTimer = new Timer(true);
        monitorTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                if (log.isInfoEnabled()) {
                    log.info("task count:" + requestSubmittedPerInterval + ",Queue size:" + scheduler.getQueue().size() + ",taskTimeConsumedPerInterval:" + taskTimeConsumedPerInterval.get() + " ms");
                }
                taskTimeConsumedPerInterval.set(0);
                requestSubmittedPerInterval.set(0);
            }
        }, 0, ONE_MINUTE_IN_SECONDS * 100);
    }

    public static RequestPersistence getInstance() {
        return InstanceHolder.INSTANCE;
    }

    @Override
    public void store(MMSDeliverRequest mmsDeliverRequest) {
        // 角色：ActiveObject.MethodRequest
        Callable<Boolean> methodRequest = new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                long start = System.currentTimeMillis();
                try {
                    delegate.store(mmsDeliverRequest);
                } finally {
                    taskTimeConsumedPerInterval.addAndGet(System.currentTimeMillis() - start);
                }
                return Boolean.TRUE;
            }
        };
        scheduler.submit(methodRequest);
        requestSubmittedPerInterval.incrementAndGet();
    }

    @Override
    public void close() throws IOException {
        scheduler.shutdown();
    }
}
